package com.eastedu.superise.scanner;

import com.eastedu.superise.CustomDataClassRowMapper;
import com.google.common.collect.Lists;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.ResultSetExtractor;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.lang.NonNull;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.LinkedList;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * The interface Callback.
 *
 * @param <T> the type parameter
 * @author superman
 */
@Slf4j
public class CustomResultSetExtractor<T> implements ResultSetExtractor<T> {
    private final long count;
    private final String className;
    private final RowMapper<T> rowMapper;

    /**
     * Instantiates a new Abstract data callback.
     */
    private final Callback<T> callback;
    private final ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
    private final Integer period = 15;
    private final Integer speed;
    private final ExecutorService executor;
    private final Semaphore semaphore;
    /**
     * 总数
     */
    private final AtomicLong total = new AtomicLong(0);
    private final AtomicLong processCount = new AtomicLong(0);
    private final LinkedList<Long> processNumbers = Lists.newLinkedList(Lists.newArrayList(0L));

    /**
     * Instantiates a new Abstract data callback.
     *
     * @param count    the count
     * @param callback the callback
     * @param clazz    the clazz
     */
    protected CustomResultSetExtractor(long count, Callback<T> callback, Class<T> clazz) {
        this.count = count;
        this.className = clazz.getSimpleName();
        this.callback = callback;
        this.rowMapper = new CustomDataClassRowMapper<>(clazz);
        /*
          这里运行任务的线程池大小需要和信号量的大小保持一致，不然主线程走完后，任务全部堆积到任务线程池里面去了，这样就无法实现停止功能
         */
        int currentSpeed = getSpeed();
        this.speed = currentSpeed;
        this.executor = Executors.newFixedThreadPool(currentSpeed);
        this.semaphore = new Semaphore(currentSpeed);

    }

    @SneakyThrows
    @Override
    public T extractData(@NonNull ResultSet rs) throws DataAccessException {
        this.startProgressMonitor();
        // 检查任务是否取消
        while (!isCancelled() && rs.next()) {
            final T row = rowMapper.mapRow(rs, (int) total.getAndIncrement());
            semaphore.acquire();
            this.executor.execute(() -> {
                try {
                    callback.callback(row);
                } catch (SQLException e) {
                    log.error("处理数据时发生错误", e);
                    // 可以在这里添加重试逻辑或其他错误处理
                } finally {
                    processCount.getAndIncrement();
                    semaphore.release();
                }
            });
        }
        return null;
    }

    private void startProgressMonitor() {
        this.scheduledExecutor.scheduleAtFixedRate(() -> printProgress(processCount.get()), period, period, TimeUnit.SECONDS);
    }

    private boolean isCancelled() {
        return false;
    }

    private void printProgress(long processedCount) {
        try {
            BigDecimal estimatedTimeRemainingMinutes = calculateEstimatedTimeRemaining(processedCount);
            log.info("<------ 数据源[{}] 剩余数据:{}/{} 预计还需[{}]分钟 当前速度[{}] ------>", this.className, count - processedCount, count, estimatedTimeRemainingMinutes, this.speed);
            if (processedCount == count || isEnd()) {
                this.exit();
            }
        } catch (Exception e) {
            checkAndExitIfEnd();
            log.warn("进度打印异常, 上一次处理数量[{}]当前处理总数[{}]:", processNumbers.getLast(), processCount.get(), e);
        }
    }

    private Long getLastEffectiveProcessCount(long effectiveProcessCount) {
        Long last = this.processNumbers.getLast();
        this.processNumbers.addLast(effectiveProcessCount);
        return last;
    }

    private void checkAndExitIfEnd() {
        if (isEnd()) {
            this.exit();
        }
    }

    private void exit() {
        try {
            this.callback.afterCompletion();
        } finally {
            this.scheduledExecutor.shutdown();
        }
    }

    private boolean isEnd() {
        return this.semaphore.availablePermits() == this.speed && !this.semaphore.hasQueuedThreads();
    }

    private BigDecimal calculateEstimatedTimeRemaining(long effectiveProcessCount) {
        Long lastProcessCount = getLastEffectiveProcessCount(effectiveProcessCount);
        if (0 == effectiveProcessCount || lastProcessCount == effectiveProcessCount) {
            return BigDecimal.valueOf(-1);
        }
        // 使用原生类型进行中间计算
        double processed = effectiveProcessCount - lastProcessCount;
        double ratePerSecond = processed / Double.valueOf(period);
        double remaining = count - effectiveProcessCount;

        // 最终步骤使用BigDecimal确保精度
        BigDecimal ratePerSecondWithBigDecimal = BigDecimal.valueOf(ratePerSecond);
        BigDecimal remainingWithBigDecimal = BigDecimal.valueOf(remaining);
        return remainingWithBigDecimal.divide(ratePerSecondWithBigDecimal, 2, RoundingMode.HALF_UP).divide(BigDecimal.valueOf(60), 2, RoundingMode.HALF_UP);
    }

    private int getSpeed() {
        return 1;
    }
}
